package com.sunday.canal.adapter.mongo.service;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.mongodb.bulk.BulkWriteResult;
import com.sunday.canal.adapter.mongo.support.EventData;
import com.sunday.canal.adapter.mongo.support.JdbcTypeUtil;
import lombok.extern.slf4j.Slf4j;
import org.bson.Document;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.BulkOperationException;
import org.springframework.data.mongodb.core.BulkOperations;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.BasicQuery;
import org.springframework.data.mongodb.core.query.BasicUpdate;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Description: mongo 同步
 * @Author: zsy
 * @Create: 2023/4/11 18:09
 */
@Slf4j
@Component
public class MongoSyncService extends BaseMongoSyncService implements InitializingBean {

    @Autowired
    private MongoTemplate mongoTemplate;

    private static MongoSyncService mongoSyncService = null;

    public static MongoSyncService getInstances() {
        return mongoSyncService;
    }

    @Override
    public void afterPropertiesSet() {
        mongoSyncService = this;
    }

    public void sync(List<Entry> entries) {
        StopWatch sw = new StopWatch("sync");
        sw.start("parseEntry");
        Collection<EventData> dataCollection = parseEntry(entries);
        sw.stop();
        sw.start("tableGroup");
        Map<String, List<EventData>> tableGroup = dataCollection.stream().collect(Collectors.groupingBy(EventData::getTableName));
        sw.stop();
        if (!dataCollection.isEmpty()) {
            try {
                sw.start("bulkOps");
                bulkOps(tableGroup);
                sw.stop();
            } catch (BulkOperationException e) {
                log.error("bulkOps error : {}", e.getMessage(), e);
                sw.stop();
                // 发生异常准备再次进行备选方案执行
                sw.start("alternativeOffer");
                alternativeOffer(tableGroup);
                sw.stop();
            }
        }
        log.info(sw.prettyPrint());
    }

    protected void alternativeOffer(Map<String, List<EventData>> tableGroup) {
        tableGroup.forEach((collection, eventDataList) -> {
            BulkOperations bulkOperations = mongoTemplate.bulkOps(BulkOperations.BulkMode.ORDERED, collection);
            for (EventData data : eventDataList) {
                if (data.getEventType() == CanalEntry.EventType.DELETE) {
                    bulkOperations.remove(new BasicQuery(data.getQuery()));
                } else if (data.getEventType() == CanalEntry.EventType.INSERT) {
                    bulkOperations.upsert(new BasicQuery(data.getQuery()), new BasicUpdate(new Document("$set", data.getDocuments())));
                } else {
                    bulkOperations.updateOne(new BasicQuery(data.getQuery()), new BasicUpdate(new Document("$set", data.getDocuments())));
                }
            }
            BulkWriteResult executeResult = bulkOperations.execute();
            log.info("[alternative] table {} bulk command execute -> insertedCount={},matchedCount={},removedCount={},modifiedCount={}",
                    collection,
                    executeResult.getInsertedCount(),
                    executeResult.getMatchedCount(),
                    executeResult.getDeletedCount(),
                    executeResult.getModifiedCount()
            );
        });
    }

    protected void bulkOps(Map<String, List<EventData>> tableGroup) {
        tableGroup.forEach((collection, eventDataList) -> {
            BulkOperations bulkOperations = mongoTemplate.bulkOps(BulkOperations.BulkMode.ORDERED, collection);
            for (EventData data : eventDataList) {
                if (data.getEventType() == CanalEntry.EventType.DELETE) {
                    bulkOperations.remove(new BasicQuery(data.getQuery()));
                } else if (data.getEventType() == CanalEntry.EventType.INSERT) {
                    bulkOperations.insert(data.merge());
                } else {
                    bulkOperations.updateOne(new BasicQuery(data.getQuery()), new BasicUpdate(new Document("$set", data.getDocuments())));
                }
            }
            BulkWriteResult executeResult = bulkOperations.execute();
            log.info("[bulkOps] table [{}] bulk command execute -> insertedCount={},matchedCount={},removedCount={},modifiedCount={}",
                    collection,
                    executeResult.getInsertedCount(),
                    executeResult.getMatchedCount(),
                    executeResult.getDeletedCount(),
                    executeResult.getModifiedCount()
            );
        });
    }


    protected EventData buildData(String tableName, CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
        Document query = new Document();
        Document documents = new Document();
        if (eventType == CanalEntry.EventType.DELETE) {
//            printColumn(rowData.getBeforeColumnsList());
            rowData.getBeforeColumnsList().stream()
                    .filter(CanalEntry.Column::getIsKey)
                    .forEach(column -> query.append("_id", typeConvert(tableName, column)));
        } else {
//            printColumn(rowData.getAfterColumnsList());
            rowData.getAfterColumnsList().stream()
                    .forEach(column -> {
                        if (column.getIsKey()) {
                            query.append("_id", typeConvert(tableName, column));
                        } else {
                            documents.append(column.getName(), typeConvert(tableName, column));
                        }
                    });
        }
        return EventData.builder()
                .tableName(tableName)
                .eventType(eventType)
                .query(query)
                .documents(documents)
                .build();
    }

    protected Object typeConvert(String tableName, Column column) {
        return JdbcTypeUtil.typeConvert(tableName, column.getName(), column.getValue(), column.getSqlType(), column.getMysqlType());
    }


}
